﻿using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;

namespace LucyGrid.Master
{
	public class ChunkBucket
	{
		private BlockingCollection<ChunkInfo> InBucket;
		private BlockingCollection<ChunkInfo> OutBucket;
		private ConcurrentDictionary<int, ChunkInfo> ProcessingBucket;
		public readonly GridMetrics Metrics;

		private class ChunksPerSecond
		{
			public long LastValue;
			public int CurrentSecond;
			public long Count;

			public void Set()
			{
				var key = DateTime.Now.Second;
				if (CurrentSecond == key)
				{
					Count++;
				}
				else
				{
					LastValue = Count;
					Count = 1;
					CurrentSecond = key;
				}
			}
		}

		private ChunksPerSecond InChunksPerSecondCounter;
		private ChunksPerSecond OutChunksPerSecondCounter;

		private bool BucketClosed;

		public ChunkBucket()
		{
			BucketClosed = false;
			InBucket = new BlockingCollection<ChunkInfo>(new ConcurrentQueue<ChunkInfo>());
			OutBucket = new BlockingCollection<ChunkInfo>(new ConcurrentQueue<ChunkInfo>());
			ProcessingBucket = new ConcurrentDictionary<int, ChunkInfo>();
			InChunksPerSecondCounter = new ChunksPerSecond();
			OutChunksPerSecondCounter = new ChunksPerSecond();
			Metrics = new GridMetrics();
		}
		/// <summary>
		/// Chunks pending to be processed count
		/// </summary>
		public long InButcketCount
		{

			get { return InBucket.Count; }
		}
		/// <summary>
		/// Add new chunk to be processed
		/// </summary>
		/// <param name="id"></param>
		/// <param name="data"></param>
		/// <returns></returns>
		public ChunkInfo Add(int id, object data)
		{
			try
			{
				InChunksPerSecondCounter.Set();
				var chunk = new ChunkInfo(id, data);
				Add(chunk);
				Metrics.TotalChunkCount++;
				return chunk;
			}
			catch (Exception ex)
			{
				throw new Exception("FATAL: Error adding new chunks to the process queue. " + ex.Message);

			}
			return null;
		}

		private void Add(ChunkInfo chunk)
		{
			chunk.Chunk.Trace("CreateChunk");
			InBucket.Add(chunk);
			UpdateMetrics();
		}

		private void UpdateMetrics(NodeInfo ninfo = null)
		{
			Metrics.PendingCount = InBucket.Count;
			Metrics.ProcessingCount = ProcessingBucket.Count;
			if (ninfo != null && Metrics.NodesConnected.All(n => n != ninfo.Id))
				Metrics.NodesConnected.Add(ninfo.Id);
			Metrics.InChunksPerSecond = InChunksPerSecondCounter.LastValue;
			Metrics.OutChunksPerSecond = OutChunksPerSecondCounter.LastValue;
		}

		/// <summary>
		/// Get next chunk to process.
		/// <remarks>If all chunks are beign processed, the thread will be blocked until:
		/// a) all are relased or 
		/// b) there's a new chunk to process</remarks>
		/// </summary>
		/// <returns></returns>
		public ChunkInfo GetNextChunk(NodeInfo ninfo)
		{
			ChunkInfo chunk = null;
			// No more chunks
			if (InBucket.IsCompleted) return chunk;
			try
			{
				// TODO: try to use a cancellation token and signal before finalizing bucket
				chunk = InBucket.Take();
				// Set chunk info
				chunk.Node = ninfo;
				chunk.StartProcessing = DateTime.Now;
				chunk.Chunk.Trace("GetNextChunk");
				// Move to processing
				ProcessingBucket.TryAdd(chunk.Id, chunk);
				// Update metrics
				UpdateMetrics();
			}
			catch (InvalidOperationException ex)
			{
				// Queue empty!
			}
			return chunk;
		}

		/// <summary>
		/// Error processing a chunk. Reset flags to be ready to be processed by another node.
		/// </summary>
		/// <param name="chunkid"></param>
		public ChunkInfo ResetChunkProcessing(int chunkid)
		{
			ChunkInfo chunk = null;
			ProcessingBucket.TryRemove(chunkid, out chunk);
			chunk.Node = null;
			chunk.Chunk.Trace("ResetChunkProcessing");
			// Back to the queue
			Add(chunk);
			return chunk;
		}

		/// <summary>
		/// Rest all chunks associated to a node
		/// </summary>
		/// <param name="node"></param>
		public void ResetChunkProcessing(NodeInfo node)
		{
			var chunks = ProcessingBucket.Values.Where(v => v.Node == node).ToArray();
			foreach (var chunk in chunks)
				ResetChunkProcessing(chunk.Id);
		}

		/// <summary>
		/// Mark chunk as finalized and ready to be returned to the main process
		/// </summary>
		/// <param name="processedChunk"></param>
		/// <param name="processTime"></param>
		/// <returns></returns>
		public ChunkInfo FinalizeChunk(Chunk processedChunk)
		{
			ChunkInfo chunk = null;
			ProcessingBucket.TryRemove(processedChunk.Head.Id, out chunk);
			chunk.Chunk = processedChunk;
			chunk.Chunk.Trace("FinalizeChunk");
			// Add to the output
			OutBucket.Add(chunk);
			// Counter
			OutChunksPerSecondCounter.Set();
			// Update metrics
			UpdateMetrics();
			Metrics.ProcessedCount++;
			// Close input bucket and release blocks if nothing else is pending
			TryCloseBucket();
			return chunk;
		}

		/// <summary>
		/// Returns next Chunk or null if there are no more chunks to process
		/// </summary>
		/// <returns></returns>
		public ChunkInfo Take()
		{
			try
			{
				return OutBucket.Take();
			}
			catch (InvalidOperationException ex)
			{
				// Empty queue
				if (ChunksInTheBucket != 0)
					throw ex;
			}
			return null;
		}
		/// <summary>
		/// Indicates if there are chunks on any bucket
		/// </summary>
		public int ChunksInTheBucket
		{
			get
			{
				return InBucket.Count +
					   ProcessingBucket.Count;
			}
		}
		/// <summary>
		/// Close bucket
		/// </summary>
		public void CompleteAdding()
		{
			BucketClosed = true;
			TryCloseBucket();
		}

		private void TryCloseBucket()
		{
			if (BucketClosed && 
				ProcessingBucket.Count == 0 &&
				InBucket.Count == 0)
			{
				InBucket.CompleteAdding();
				OutBucket.CompleteAdding();
			}
		}
		
	}
}
